package kafka

import (
	"fmt"
	"github.com/Shopify/sarama"
	cluster "github.com/bsm/sarama-cluster"
	"github.com/zeromicro/go-zero/core/logx"
	"go.uber.org/zap"
	"time"
)

type operatorKafkaMsg func(*sarama.ConsumerMessage) error

type Consumer struct {
	addrs     []string
	groupID   string
	topicList []string
	config    *cluster.Config
	consumer  *cluster.Consumer
	operator  operatorKafkaMsg
}

// NewDefaultConsumer 初始化消费者
func NewDefaultConsumer(groupid string, addrs, topicList []string, operator operatorKafkaMsg) *Consumer {
	config := cluster.NewConfig()
	// 消费消息出现错误时,发送error
	config.Consumer.Return.Errors = true
	// 当发生rebalance时,通知信息
	config.Group.Return.Notifications = true
	config.Consumer.Offsets.CommitInterval = 1 * time.Second
	// 从老消息开始消费
	config.Consumer.Offsets.Initial = sarama.OffsetOldest
	if operator == nil {
		panic("kafka consumer operator can not be nil")
	}
	return &Consumer{
		addrs:     addrs,
		groupID:   groupid,
		topicList: topicList,
		config:    config,
		operator:  operator,
	}
}

// Start kafka consumer
func (c *Consumer) Start() (err error) {
	c.consumer, err = cluster.NewConsumer(c.addrs, c.groupID, c.topicList, c.config)
	if err != nil {
		//log.Get().Error("start Consumer Error", zap.Error(err))
		return
	}
	if c.operator == nil {
		panic("kafka consumer operator can not be nil")
	}
	go func() {
		for err = range c.consumer.Errors() {
			logx.Error("Consumer Error", zap.Time("time", time.Now()), zap.Error(err))
			fmt.Printf("Consumer Error:%v", err)
		}
	}()

	go func() {
		for note := range c.consumer.Notifications() {
			// 或许需要发送通知
			logx.Info("GetConsumerNotifications", zap.Time("time", time.Now()), zap.Reflect("Rebalanced", note))
		}
	}()
	go func() {
		for msg := range c.consumer.Messages() {
			if err = c.operator(msg); err != nil {
				time.Sleep(10 * time.Millisecond)
				continue
			}
			// 手动提交offset
			c.consumer.MarkOffset(msg, "")
		}
	}()
	return
}
